/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package miner

import (
	"sync"

	"github.com/golang/protobuf/proto"
	"go.uber.org/atomic"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/message"
	"hundsun.com/hsl/hschain/common/types"
	"hundsun.com/hsl/hschain/common/util/pb"
	consensuscomm "hundsun.com/hsl/hschain/consensus/common"
	dposcomm "hundsun.com/hsl/hschain/consensus/impl/dpos/common"
	consensustypes "hundsun.com/hsl/hschain/consensus/types"
	"hundsun.com/hsl/hschain/protos"
	"hundsun.com/hsl/hschain/protos/common"
	pbdpos "hundsun.com/hsl/hschain/protos/consensus/dpos"
	pbledger "hundsun.com/hsl/hschain/protos/ledger"
	"hundsun.com/hsl/hschain/store/mq"
)

// NewEdBlockBuffer 新建一个EdBlocksBuffer对象
func NewEdBlockBuffer(client mq.Client, log logging.Log) *EdBlocksBuffer {
	return &EdBlocksBuffer{
		data:        make([]*EdBlock, dposcomm.MaxBlocksInTimeSlice),
		dataSize:    atomic.NewInt32(0),
		topicClient: client,
		log:         log,
	}
}

// EdBlocksBuffer 用于存放本时间片内生成的块及背书结果
type EdBlocksBuffer struct {
	data        []*EdBlock
	dataSize    *atomic.Int32
	topicClient mq.Client
	seqClientMu sync.Mutex
	log         logging.Log
}

// Put 放入一个BlockEndorsements对象
func (e *EdBlocksBuffer) Put(ed *EdBlock) {
	dataSize := e.dataSize.Load()
	if dataSize == int32(dposcomm.MaxBlocksInTimeSlice) {
		e.log.Error("EdBlocksBuffer is full")
		return
	}

	e.data[dataSize] = ed
	e.dataSize.Inc()
}

// Get 根据区块哈希获取BlockEndorsements对象
func (e *EdBlocksBuffer) Get(blockNumber uint64) *EdBlock {
	for _, edBlk := range e.data {
		if edBlk.block.Header.Number == blockNumber {
			return edBlk
		}
	}
	return nil
}

//Clear 清空数据，重置状态
func (e *EdBlocksBuffer) Clear() {
	e.dataSize.Store(0)
}

//TryToSendPassedBlocksToSeqCenter 尝试将背书通过的块发送到序列号中心，必须按照顺序提交
func (e *EdBlocksBuffer) TryToSendPassedBlocksToSeqCenter(seqCenterPeerID string, blockQueue chan<- *consensuscomm.BlockAssign, bSequenceServerLocal bool) {
	e.seqClientMu.Lock()
	defer e.seqClientMu.Unlock()
	dataSize := int(e.dataSize.Load())
	for i := 0; i < dataSize; i++ {
		edBlock := e.data[i]
		if edBlock.status != EdBlockStatusPassed {
			return
		}
		if !edBlock.hasBeenSentToSeqCenter {
			endorsedBlock := edBlock.ConvertToEndorsedBlock()
			pb.GetBlockLog(e.log, edBlock.block).Infof("begin to write block to sequence center: %s", seqCenterPeerID)
			if bSequenceServerLocal {
				blockQueue <- &consensuscomm.BlockAssign{Block: endorsedBlock, OriginPeerID: ""}
			} else {
				err := e.WriteBlockSequenceBlock(seqCenterPeerID, endorsedBlock)
				if err != nil {
					pb.GetBlockLog(e.log, edBlock.block).Debug("write block to sequence center failed, err:%s\n", err.Error())
					return
				}
			}
			edBlock.hasBeenSentToSeqCenter = true
			pb.GetBlockLog(e.log, edBlock.block).Debug("put into send queue")
		}
	}
}

// WriteBlockSequenceBlock 将区块发送到blockSequence进行编号
func (e *EdBlocksBuffer) WriteBlockSequenceBlock(seqCenterPeerID string, block *pbledger.Block) error {
	var relayMsg = &consensustypes.ConsensRelayMsg{}
	relayMsg.MsgType = consensustypes.BlockSequenceAddBlockMsg
	payload, err := proto.Marshal(block)
	if err != nil {
		return err
	}
	relayMsg.Payload = payload

	p2pMsg, err := message.EncodeMessage(seqCenterPeerID, relayMsg)
	if err != nil {
		return err
	}
	addSeqBlkMsg := e.topicClient.NewMessage(types.P2pModule, mq.ModeTopic.String(), types.EventConsensusRequestMsg, p2pMsg)
	err = e.topicClient.Send(addSeqBlkMsg, false)
	if err != nil {
		return err
	}
	return nil
}

// NewEdBlock 新建一个EdBlock对象
func NewEdBlock(block *pbledger.Block, currentMiners map[protos.Address]*common.Miner, log logging.Log) *EdBlock {
	return &EdBlock{
		block:               block,
		status:              EdBlockStatusUnknown,
		currentMiners:       currentMiners,
		rejectMiners:        make(map[protos.Address]interface{}, dposcomm.MaxMinersCount),
		approveEndorsements: make(map[protos.Address]*pbdpos.BlockEndorsementResponse, dposcomm.MaxMinersCount),
		log:                 log,
	}
}

// BlockEdStatus 区块背书状态
type BlockEdStatus int

const (
	// EdBlockStatusUnknown 区块背书状态--未知（还未收到足够多请求）
	EdBlockStatusUnknown = iota

	// EdBlockStatusPassed 区块背书状态--通过
	EdBlockStatusPassed

	// EdBlockStatusFailed 区块背书状态--未通过
	EdBlockStatusFailed
)

// EdBlock 已请求背书的块
type EdBlock struct {
	block                  *pbledger.Block
	status                 BlockEdStatus
	currentMiners          map[protos.Address]*common.Miner
	rejectMiners           map[protos.Address]interface{}
	approveEndorsements    map[protos.Address]*pbdpos.BlockEndorsementResponse
	log                    logging.Log
	hasBeenSentToSeqCenter bool
	sync.RWMutex
}

// Put 增加一个背书并返回区块背书状态
func (b *EdBlock) Put(edResp *pbdpos.BlockEndorsementResponse) (_ BlockEdStatus, firstAchieved bool) {
	bLog := b.log.WithField(logging.LogFieldBlockNumber, edResp.BlockNumber)
	if !edResp.Verify() {
		bLog.Warn("failed to verify ed signature")
		b.RLock()
		defer b.RUnlock()
		return b.status, false
	}
	endorser := edResp.Signature.GetAddress()
	b.Lock()
	defer b.Unlock()
	if _, ok := b.currentMiners[endorser]; !ok {
		bLog.Warnf("endorser '%s' is not miner of that round", endorser)
		return b.status, false
	}
	if _, ok := b.rejectMiners[endorser]; ok {
		bLog.Warnf("ed from '%s' exists in  rejectMiners", endorser)
		return b.status, false
	}
	if _, ok := b.approveEndorsements[endorser]; ok {
		bLog.Warnf("ed from '%s' exists in  approveEndorsements", endorser)
		return b.status, false
	}
	if edResp.Approved {
		b.approveEndorsements[endorser] = edResp
		bLog.Tracef("got approved ed from '%s', current approved:%d", endorser, len(b.approveEndorsements))
		if len(b.approveEndorsements) >= dposcomm.MinApproveMinerCount {
			b.status = EdBlockStatusPassed
			firstAchieved = len(b.approveEndorsements) == dposcomm.MinApproveMinerCount
		}
	} else {
		b.rejectMiners[endorser] = struct{}{}
		bLog.Infof("got rejected ed from '%s', current rejected:%d,errMsg:%s", endorser, len(b.rejectMiners), edResp.ErrorMessage)
		if len(b.rejectMiners) >= dposcomm.MaxRejectMinerCount {
			b.status = EdBlockStatusFailed
			firstAchieved = len(b.rejectMiners) == dposcomm.MaxRejectMinerCount
		}
	}

	return b.status, firstAchieved
}

// ConvertToEndorsedBlock 将此对象转换成一个待背书签名的块
func (b *EdBlock) ConvertToEndorsedBlock() *pbledger.Block {
	b.RLock()
	defer b.RUnlock()
	var approvedEds []*pbdpos.BlockEndorsementResponse
	for _, v := range b.approveEndorsements {
		approvedEds = append(approvedEds, v)
	}
	responses := &pbdpos.BlockEndorsementResponses{
		Responses: approvedEds,
	}
	payload, err := proto.Marshal(responses)
	if err != nil {
		panic(err)
	}
	b.block.Header.Consensus = payload
	return b.block
}
